package mqtt;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.hawtdispatch.Dispatch;
import org.fusesource.mqtt.client.*;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Properties;

@SuppressWarnings("deprecation")
public class MqttTest {

	private static int count = 0;

	private static int connectedCount = 0;
	private static Properties props = new Properties();
	private static Producer<String, String> producer = null;

	static {
		props.put("bootstrap.servers", "172.18.84.143:9092");
		props.put("acks", "0");
		props.put("retries", "0");
		props.put("batch.size", "16384");
		props.put("linger.ms", "1");
		props.put("buffer.memory", "33554432");
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

		// 创建一个生产者对象
		producer = new KafkaProducer<String, String>(props);
	}

	private String host;

	{
		try {
			host = InetAddress.getLocalHost().getHostAddress();
		} catch (UnknownHostException e) {
			e.printStackTrace();
			System.exit(1);
		}
	}

	public synchronized static void countUp() {
		connectedCount++;
	}

	public synchronized static void countDown() {
		connectedCount--;
	}

	public static void main(String[] args) throws Exception {

		String clientHost = null;
		int connectCount = 0;
		String sercerHostPort = null;

		if (args.length == 2) {
			InetAddress address = InetAddress.getLocalHost();

			clientHost = address.getHostAddress();
			connectCount = Integer.parseInt(args[0]);
			sercerHostPort = args[1];
		} else if (args.length == 3) {

			clientHost = args[0];
			connectCount = Integer.parseInt(args[1]);
			sercerHostPort = args[2];
		} else {

			System.out.println("请检查输入参数是否正确");
			System.out.println("参数格式：");
			System.out.println("	java -jar xxx.jar clientHost connectCount sercerHostPort");
			System.out.println("或：");
			System.out.println("	java -jar xxx.jar connectCount sercerHostPort");
			System.out.println("示例：");
			System.out.println("	java -jar xxx.jar 172.18.191.90 30000 172.18.166.117:7070");
			System.out.println("	java -jar xxx.jar 30000 172.18.166.117:7070");
			System.exit(1);
		}

//		String clientHost = "172.18.191.90";
//		int connectCount = 5;
//		String sercerHostPort = "172.18.166.117:7070";

		for (int i = 0; i < connectCount; ) {
			i++;
			new MqttTest().runMqttTest(sercerHostPort, clientHost, i);

//			if (i % 10000 == 0) {
//				try {
//					Thread.sleep(5000);
//				} catch (InterruptedException e) {
//					// TODO Auto-generated catch block
//					e.printStackTrace();
//				}
//			}
		}

		while (true) {

			System.out.println("已建立连接数：" + MqttTest.connectedCount + "	QPS：" + (MqttTest.count / 10));
			MqttTest.count = 0;

			try {
				Thread.sleep(1000 * 10);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * @param sercerHostPort 172.18.166.117:7070
	 * @param clientHost     172.18.191.90_1_2
	 * @param count
	 */
	public void runMqttTest(String sercerHostPort, String clientHost, int count) {

		try {

			MQTT mqtt = new MQTT();

			mqtt.setHost("tcp://" + sercerHostPort);

			mqtt.setClientId(clientHost + "_" + count + "_2"); // 用于设置客户端会话的ID。在setCleanSession(false);被调用时，MQTT服务器利用该ID获得相应的会话。此ID应少于23个字符，默认根据本机地址、端口和时间自动生成
			mqtt.setCleanSession(false); // 若设为false，MQTT服务器将持久化客户端会话的主体订阅和ACK位置，默认为true
			mqtt.setKeepAlive((short) 120);// 定义客户端传来消息的最大时间间隔秒数，服务器可以据此判断与客户端的连接是否已经断开，从而避免TCP/IP超时的长时间等待
			mqtt.setUserName(clientHost + "_" + count + "_2");// 服务器认证用户名
			mqtt.setPassword("admin");// 服务器认证密码

			mqtt.setWillTopic("willTopic");// 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息
			mqtt.setWillMessage("willMessage");// 设置“遗嘱”消息的内容，默认是长度为零的消息
			mqtt.setWillQos(QoS.AT_LEAST_ONCE);// 设置“遗嘱”消息的QoS，默认为QoS.ATMOSTONCE
			mqtt.setWillRetain(true);// 若想要在发布“遗嘱”消息时拥有retain选项，则为true
			mqtt.setVersion("3.1");

			// 失败重连接设置说明
			mqtt.setConnectAttemptsMax(-1);// 客户端首次连接到服务器时，连接的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1
			mqtt.setReconnectAttemptsMax(-1);// 客户端已经连接到服务器，但因某种原因连接断开时的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1
			mqtt.setReconnectDelay(1000L);// 首次重连接间隔毫秒数，默认为10ms
			mqtt.setReconnectDelayMax(30000L);// 重连接间隔毫秒数，默认为30000ms
			mqtt.setReconnectBackOffMultiplier(2);// 设置重连接指数回归。设置为1则停用指数回归，默认为2

			// Socket设置说明
			mqtt.setReceiveBufferSize(65536);// 设置socket接收缓冲区大小，默认为65536（64k）
			mqtt.setSendBufferSize(65536);// 设置socket发送缓冲区大小，默认为65536（64k）
			mqtt.setTrafficClass(8);// 设置发送数据包头的流量类型或服务类型字段，默认为8，意为吞吐量最大化传输

			// 带宽限制设置说明
			mqtt.setMaxReadRate(0);// 设置连接的最大接收速率，单位为bytes/s。默认为0，即无限制
			mqtt.setMaxWriteRate(0);// 设置连接的最大发送速率，单位为bytes/s。默认为0，即无限制

			// 选择消息分发队列
			mqtt.setDispatchQueue(Dispatch.createQueue("MqttDispatchQueue"));// 若没有调用方法setDispatchQueue，客户端将为连接新建一个队列。如果想实现多个连接使用公用的队列，显式地指定队列是一个非常方便的实现方法

			final CallbackConnection connection = mqtt.callbackConnection();

			connection.listener(new Listener() {

				@Override
				public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) {

					long current = System.currentTimeMillis();

					try {
						ByteArrayOutputStream baos = new ByteArrayOutputStream();
						payload.writeTo(baos);
						JSONObject responseJson = JSONObject.parseObject(baos.toString());
						// System.out.println(baos.toString());

						JSONObject resultJson = new JSONObject();
						resultJson.put("elapsed", (current - responseJson.getJSONObject("data").getLong("timeStamp")));
						resultJson.put("allThreads", connectedCount);
						resultJson.put("host", host);

						MqttTest.count++;

						// 发送消息
						producer.send(new ProducerRecord<String, String>("test", "MQTT Sub Sampler", resultJson.toJSONString()));

						ack.run();
					} catch (IOException e) {
						e.printStackTrace();
						System.exit(1);
					}

					ack.run();
				}

				@Override
				public void onFailure(Throwable value) {
					// System.out.println("onFailure");

				}

				@Override
				public void onDisconnected() {
					// System.out.println("onDisconnected");
					countDown();
				}

				@Override
				public void onConnected() {
					// System.out.println("onConnected");
					countUp();
				}
			});

			connection.connect(new Callback<Void>() {

				@Override
				public void onSuccess(Void v) {
					// System.out.println("onSuccess");
				}

				@Override
				public void onFailure(Throwable value) {
					// System.out.println("onFailure");
				}
			});

			Topic[] topics = {new Topic("test_topic_" + clientHost + "_" + count, QoS.AT_LEAST_ONCE)};

			// 订阅topic
			connection.subscribe(topics, new Callback<byte[]>() {

				@Override
				public void onSuccess(byte[] qoses) {
					// System.out.println("subscribe Successed");

				}

				@Override
				public void onFailure(Throwable value) {
					// System.out.println("subscribe Failured: " + value);

				}
			});
		} catch (Exception e) {
			// TODO: handle exception
		}
	}
}
